package com.fstec.ks.master.impl;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

import java.io.File;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fstec.ks.master.ServerCallback;
import com.fstec.ks.master.bean.RegistInfo;
import com.fstec.ks.master.queue.MessageQueue;
import com.fstec.ks.master.queue.QueueMessage;
import com.sinotn.SinotnLogger;
import com.sinotn.error.MessageException;
import com.sinotn.ks.definition.Message;
import com.sinotn.ks.definition.MessageFile;
import com.sinotn.ks.netty.DefaultMessageNotifier;
import com.sinotn.ks.netty.MessageHandler;
import com.sinotn.ks.netty.MessageNotifier;
import com.sinotn.ks.netty.message.DefaultMessageFile;
import com.sinotn.ks.netty.message.SimpleMessage;

public class NettyClient implements Runnable,MessageHandler{
	public static final AttributeKey<NettyClient> NETTY_CLIENT=AttributeKey.newInstance("MASTER_NETTY_CLIENT");
	public static final AttributeKey<String> LOGIN_TOKEN=AttributeKey.newInstance("MASTER_LOGIN_TOKEN");
	private ConcurrentHashMap<Long, MessageNotifier> notifiers=new ConcurrentHashMap<Long, MessageNotifier>();
	private static Logger log=LoggerFactory.getLogger("com.sinotn");
	// 服务端token为奇数，客户端为偶数
	private AtomicLong tokenCounter=new AtomicLong(2);
	private static final long tokenStep=2;
	private String host;
	private int port;
	private boolean keepAlive=false;
	private int sendBuffSize;
	private int receiveBuffSize;
	private boolean running=false;
	private int errorThreshold=100;
	private int errCount;
	private Channel channel;
	private ChannelHandler channelHandler;
	private RegistInfo registInfo;
	private MessageQueue messageQueue;
	private ServerCallback serverCallback;
	private ReentrantLock lock=new ReentrantLock();
	private ReentrantLock runLock=new ReentrantLock();

	public NettyClient(String host, int port, boolean ssl){
		this.host=host;
		this.port=port;
	}

	public void setMessageQueue(MessageQueue messageQueue) {
		this.messageQueue = messageQueue;
	}

	public void setRegistInfo(RegistInfo registInfo) {
		this.registInfo = registInfo;
	}

	public void setChannelHandler(ChannelHandler channelHandler) {
		this.channelHandler = channelHandler;
	}

	public void setKeepAlive(boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public void setSendBuffSize(int sendBuffSize) {
		this.sendBuffSize = sendBuffSize;
	}

	public void setServerCallback(ServerCallback serverCallback) {
		this.serverCallback = serverCallback;
	}

	public void setReceiveBuffSize(int receiveBuffSize) {
		this.receiveBuffSize = receiveBuffSize;
	}

	public void connect(String host, int port, boolean sync){
		if(this.host.equals(host) && this.port==port){
			if (this.channel==null || (!this.channel.isActive() && !this.keepAlive)){
				this.connect(sync);
			}
		} else {
			this.host=host;
			this.port=port;
			this.connect(sync);
		}
	}

	public boolean isKeepAlive() {
		return keepAlive;
	}
	
	/**
	 * 网络连接成功回调此方法
	 * @author <a href="mailto:xieqj@sinotn.com">谢启进</a>
	 * @date 2019年1月8日 上午10:38:11
	 */
	public void onConnectSuccess(){
		if(this.registInfo!=null && this.registInfo.getRegistId()!=null){
			this.login();
		}
	}
	/**
	 * 网络连接失败或连接断开回调此方法
	 * @author <a href="mailto:xieqj@sinotn.com">谢启进</a>
	 * @date 2019年1月8日 上午10:38:37
	 */
	public void onConnectFail(){}

	/**
	 * 创建服务器连接
	 * @param sync
	 * @auth <a href="mailto:xieqj@sinotn.com">谢启进</a>
	 * @date 2018年11月18日 下午5:49:58
	 */
	public void connect(boolean sync){
		if(this.channel!=null){
			this.channel.eventLoop().shutdownGracefully();
		}
		EventLoopGroup group = new NioEventLoopGroup();
		Bootstrap b = new Bootstrap();
		b.group(group);
		b.channel(NioSocketChannel.class);
		if(this.keepAlive){
			/*
			 是否启用心跳保活机制。在双方TCP套接字建立连接后（即都进入ESTABLISHED状态）并且在两个小时左右上层没有任何数据传输的情况下，这套机制才会被激活。
			 */
			b.option(ChannelOption.SO_KEEPALIVE, true);
		}

		/*
		 在TCP/IP协议中，无论发送多少数据，总是要在数据前面加上协议头，同时，对方接收到数据，也需要发送ACK表示确认。为了尽可能的利用网络带宽，TCP总是希望尽可能的发送足够大的数据。
		 这里就涉及到一个名为Nagle的算法，该算法的目的就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。
 		 TCP_NODELAY就是用于启用或关于Nagle算法。如果要求高实时性，有数据发送时就马上发送，就将该选项设置为true关闭Nagle算法；如果要减少发送次数减少网络交互，就设置为false等累积一定大小后再发送。
 		 默认为false。
		 */
		b.option(ChannelOption.TCP_NODELAY, true);
		if(this.sendBuffSize>0){
			// 定义传输的系统缓冲区buf的大小
			b.option(ChannelOption.SO_SNDBUF, this.sendBuffSize);
		}
		if(this.receiveBuffSize>0){
			// 定义接收的系统缓冲区buf的大小
			b.option(ChannelOption.SO_RCVBUF  , this.receiveBuffSize);
		}
		b.handler(this.channelHandler);
		ChannelFuture f=b.connect(host, port);
		
		if (sync){
			// 同步创建连接
			try{
				this.channel=f.sync().channel();
				if (log.isDebugEnabled()){
					log.debug(this.channel.id().asLongText() + " isActive " +this.channel.isActive());
				}
				this.channel.attr(NETTY_CLIENT).set(this);
			}catch(Throwable e){
				log.error("创建服务器Netty连接失败", e);
				group.shutdownGracefully();
				throw new MessageException("无法创建服务器连接[host="+this.host+";port="+this.port+"]");
			}
		} else {
			// 异常创建连接
			if(this.keepAlive){
				// 连接失败重连机制
				ConnectListener listener=new ConnectListener(this);
				f.addListener(listener);
			}
			this.channel=f.channel();
			this.channel.attr(NETTY_CLIENT).set(this);
			if (log.isDebugEnabled()){
				log.debug(this.channel.id().asLongText() + " isActive " +this.channel.isActive());
			}
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	protected void shutdown(GenericFutureListener listener){
		if(this.channel!=null){
			if(log.isDebugEnabled()){
				log.debug("关闭Socket连接"+this.channel.id().asShortText());
			}
			Future<?> f=this.channel.eventLoop().shutdownGracefully();
			if(listener!=null){
				f.addListener(listener);
			}
			this.channel=null;
		}
	}

	/**
	 * 判断当前客户端是否已连接
	 * @return
	 * @auth <a href="mailto:xieqj@sinotn.com">谢启进</a>
	 * @date 2018年11月19日 下午3:32:32
	 */
	public boolean isConnected() {
		if (this.channel==null){
			return false;
		}
		return this.channel.isActive();
	}

	public long nextToken(){
		return this.tokenCounter.getAndAdd(tokenStep);
	}

	public MessageNotifier send(Message sm) {
		if(this.channel==null || !channel.isActive()){
			throw new MessageException("当前服务连接不可用");
		}
		DefaultMessageNotifier notifier=new DefaultMessageNotifier();
		notifier.setToken(sm.getToken());
		notifiers.put(sm.getToken(), notifier);
		try{
			this.channel.writeAndFlush(sm);
			return notifier;
		}catch(Throwable e){
			this.notifiers.remove(sm.getToken());
			log.error("发送监控消息发生系统错误", e);
			throw new MessageException("发送监控消息失败");
		}
	}

	public void onRcv(Message msg) {
		// token 奇数为服务端推送消息
		long token=msg.getToken();
		boolean ok=false;
		if(((token>>1)<<1)==token){
			// 偶数，客户端请求响应消息
			MessageNotifier notifier=this.notifiers.remove(msg.getToken());
			if(notifier!=null){
				notifier.onRcv(msg);
				ok=true;
			}
		}else{
			// 奇数，平台推送消息
			if(this.serverCallback!=null){
				this.serverCallback.onRcv(msg);
				ok=true;
			}
		}
		if(!ok){
			MessageFile[] mfs=msg.getFiles();
			if(mfs!=null){
				int len=mfs.length;
				for(int i=0;i<len;i++){
					mfs[i].delete();
				}
			}
		}
	}

	/**
	 * 当前网络连接已断开
	 * @auth <a href="mailto:xieqj@sinotn.com">谢启进</a>
	 * @date 2018年11月19日 下午6:23:30
	 */
	public void onInactive() {
		if(!this.notifiers.isEmpty()){
			// 处理所有等待通知
			MessageNotifier notifier;
			long lastToken=this.tokenCounter.get() - 1;
			do{
				notifier=this.notifiers.remove(lastToken--);
				if(notifier!=null){
					notifier.onRcv(null);
				}
			}while(!this.notifiers.isEmpty());
		}
		if(this.keepAlive && this.registInfo!=null && this.registInfo.isRegist()){
			// 如果是长连接，尝试重连
			if(this.channel==null){
				this.connect(false);
			} else {
				this.channel.eventLoop().schedule(new Runnable() {
					@Override
					public void run() {
						connect(false);
					}
				}, 10L, TimeUnit.SECONDS);// 20秒后重连
			}
		}
	}

	public void onRspTimeout() {
		if(!this.notifiers.isEmpty()){
			// 一次读心跳超时，检查所有等待通知，对超时等待通知处理。
			MessageNotifier notifier;
			long lastToken=this.tokenCounter.get() - 1;
			long timeout;
			long token;
			int size=this.notifiers.size();
			while(size>0){
				token=lastToken--;
				if(token<1){
					break;
				}
				notifier=this.notifiers.get(token);
				if(notifier==null){
					continue;
				}
				size--;
				timeout=notifier.getTimeout();
				if(timeout==0 || (System.currentTimeMillis()> timeout)){
					notifier.onRcv(null);
					this.notifiers.remove(token);
				}
			}
		}
	}
	
	private void login() {
		if(this.channel==null || !this.channel.isActive()){
			return;
		}
		if (this.lock.tryLock()){
			try{
				final String serverId=this.registInfo.getRegistId();
				if(!serverId.equals(this.channel.attr(LOGIN_TOKEN).get())){
					SimpleMessage sm=new SimpleMessage(102, this.nextToken());
					sm.setText(serverId);
					MessageNotifier notifier=this.send(sm);
					notifier.setMessageHandler(this);
					/*
					ChannelFuture future=this.channel.writeAndFlush(sm);
					future.addListener(new LoginListener(serverId, this));
					*/
				}
			}finally{
				this.lock.unlock();
			}
		}
	}
	@Override
	public void handle(Message message) {
		if(message!=null){
			// 处理登录响应消息
			boolean retry=true;
			if(message.getType()==0){
				// 登录成功
				channel.attr(LOGIN_TOKEN).set(message.getText());
				this.runLock.lock();
				try{
					if(!this.running && this.errCount<this.errorThreshold){// 错误保护
						new Thread(this).start();
						this.running=true;
					}
				}finally{
					this.runLock.unlock();
				}
				retry=false;
			}else if(message.getType()==101){
				// 已经过期
				this.registInfo.setRegistId(null);
				this.registInfo.setRegistTime(null);
				retry=false;
				this.shutdown(null);
				if(this.messageQueue!=null){
					this.messageQueue.close(true);
				}
				this.registInfo.save();
			}
			if(retry){
				SinotnLogger.DEBUG.error("注册监控服务失败，1s后重试");
				final EventLoop loop = channel.eventLoop();
				loop.schedule(new Runnable() {
					@Override
					public void run() {
						login();
					}
				}, 3L, TimeUnit.SECONDS);
			}
		}
	}
	public boolean isActive(){
		if(this.channel==null) return false;
		if(!this.channel.isActive()) return false;
		if(channel.attr(LOGIN_TOKEN).get()==null) return false;
		return true;
	}
	@Override
	public void run() {
		try{
			Thread.sleep(1000);
			QueueMessage message=null;
			while(true){
				if(this.isActive() && this.messageQueue!=null){
					try{
						if(message!=null){
							this.innerSend(message);
							message=null;
						}
						for(message=this.messageQueue.next();message!=null;message=this.messageQueue.next()){
							this.innerSend(message);
							message=null;
						}
						if(this.errCount>0){
							this.errCount=0;
						}
					}catch(MessageException e){
						SinotnLogger.DEBUG.error(e.getMessage());
					}catch(Throwable e){
						SinotnLogger.DEBUG.error("处理监控消息失败pos="+message.getPosition(), e);
						this.errCount++;
						if(this.errCount>this.errorThreshold){
							break;
						}
					}
				}
				Thread.sleep(10000);
			}
		}catch(Throwable e){
			SinotnLogger.DEBUG.error("异步监控消息推送任务错误退出", e);
		}finally{
			this.runLock.lock();
			try{
				this.running=false;
			}finally{
				this.runLock.unlock();
			}
		}
	}

	private void innerSend(QueueMessage queueMessage) {
		File file=queueMessage.getFile();
		if(file!=null && !file.exists()){
			// 队列消息中指定文件如果已不存在，输出日志跳过发送(避免由于文件被人为删除导致后续消息不能发送)
			SinotnLogger.DEBUG.error("推送监控消息文件不存在"+file.getAbsolutePath());
			return;
		}
		SimpleMessage sm=new SimpleMessage(queueMessage.getType(), this.nextToken());
		if(queueMessage.getBodyText()!=null){
			sm.setText(queueMessage.getBodyText());
		}
		if(queueMessage.getFile()!=null){
			sm.setMessageFiles(new MessageFile[]{
				new DefaultMessageFile(queueMessage.getFile())
			});
		}
		MessageNotifier notifier=this.send(sm);
		notifier.doWait(0, "发送异步监控消息");
		Message message=notifier.getResponse();
		if(message==null){
			throw new MessageException("推送["+queueMessage.getType()+"]监控消息响应null"+sm.toString());
		}
		if(message.getType()!=0){
			throw new MessageException("推送["+queueMessage.getType()+"]监控消息响应异常("+message.getText()+")"+sm.toString());
		}
	}
}
